
//
// test Insert-Append operation
//

#include <stdio.h>
#include <string.h>

#include <sstream>
#include <iostream>
#include <iomanip>

#include <boost/bind.hpp>
#include <boost/thread/thread.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>

#include <boost/filesystem.hpp>

#include <boost/program_options.hpp>
namespace po = boost::program_options;

using namespace boost::posix_time;

#include "sancab/client.hpp"

#define	N_ROUND	10000

float
getTimeDuration( ptime time_begin, ptime time_end )
{
    time_period tp( time_begin, time_end );
    time_duration   td = tp.length();
    long    time_last_ms = td.total_milliseconds();
    float   time_last_s = time_last_ms / 1000.0f;

    return time_last_s;
}

const char*	DBT_PREFIX = "InsertAppend-";

const int	NUMBER_VALUES_PER_KEY = 5;

void
testfunc( const std::string& appconf, int times, bool verbose, int table_id )
{
	char	table_name[ 32 ];
	snprintf(table_name, sizeof(table_name) - 1, "cltTest%09u", table_id );

	sancab::client::Client	clt( appconf.c_str() );

	{
		ptime   time_start = microsec_clock::local_time();
		ptime	time_start_R = time_start;
		u_int32_t	count_down_R = N_ROUND;

		for (int i=0; i<times; i++) {
			char	key[32];
			char	value[32];

			snprintf( key, sizeof(key) - 1, "%s-KEY-%08d", DBT_PREFIX, i+1 );
            std::string _key( key );

			// insert some values to APPEND
			for (int k=0; k < NUMBER_VALUES_PER_KEY; ++k) {
				snprintf( value, sizeof(value) - 1, "%s-DATA-%08d--APPEND-%08d", DBT_PREFIX, i+1, k+1 );
				std::string _value( value );

				int rc = clt.insertAppend( _key, _value );
				if (rc < 0) {
					std::ostringstream	msg;
					msg << "fail to PUT-Append(table:" << table_name << ", key:" << key << ", value:" << value << ")";
					throw std::runtime_error( msg.str() );
				}
			}

			--count_down_R;
			if (0 == count_down_R) {
				ptime   time_stop_R = microsec_clock::local_time();
				float   time_last_s = getTimeDuration( time_start_R, time_stop_R );
				float   tps = N_ROUND / time_last_s;
				std::cerr << "  R-PUT " << N_ROUND << " records last " << std::setprecision(3) << std::fixed << time_last_s << " seconds, tps=" << tps << std::resetiosflags(std::ios::floatfield) << std::setprecision(6) << std::endl;

				time_start_R = time_stop_R;
				count_down_R = N_ROUND;
			}
		}

		ptime   time_stop = microsec_clock::local_time();
		float   time_last_s = getTimeDuration( time_start, time_stop );
		float   tps = times / time_last_s;
		std::cerr << "PUT " << times << " records last " << std::setprecision(3) << std::fixed << time_last_s << " seconds, tps=" << tps << std::resetiosflags(std::ios::floatfield) << std::setprecision(6) << std::endl;
	}


	{
		ptime   time_start = microsec_clock::local_time();
		ptime	time_start_R = time_start;
		u_int32_t	count_down_R = N_ROUND;

		for (int i=0; i<times; i++) {
			char	key[32];
			char	value[32];

			snprintf( key, sizeof(key) - 1, "%s-KEY-%08d", DBT_PREFIX, i+1 );
			snprintf( value, sizeof(value) - 1, "%s-DATA-%08d", DBT_PREFIX, i+1 );

            std::string _key( key );

			for (int k=1; k <= NUMBER_VALUES_PER_KEY; ++k) {
				std::vector< std::string > buf_v;
				u_int32_t	limit = k;

				int rc = clt.find( _key, buf_v, limit );
				if (rc < 0) {
					std::ostringstream	msg;
					msg << "fail to GET(table:" << table_name << ", key:" << key << ")";
					throw std::runtime_error( msg.str() );
				}
				else if (rc == 0) {
					std::ostringstream	msg;
					msg << "NOT-FOUND GET(table:" << table_name << ", key:" << key << ")";
					throw std::runtime_error( msg.str() );
				}
				else if ( k != buf_v.size() ) {
					std::ostringstream	msg;
					msg << "GET(table:" << table_name << ", key:" << key << ") -- GOT " << buf_v.size() << " records, but WANT " << k << " ones.";
					throw std::runtime_error( msg.str() );
				}

				for (int n=0; n<k; ++n) {
					snprintf( value, sizeof(value) - 1, "%s-DATA-%08d--APPEND-%08d", DBT_PREFIX, i+1, NUMBER_VALUES_PER_KEY - n );
					std::string want_value( value );
					const std::string&	got_value = buf_v[ n ];
					if ( want_value != got_value ) {
						std::ostringstream	msg;
						msg << "GET(table:" << table_name << ", key:" << key << ") -- GOT:" << got_value << " WANT:" << want_value;
						throw std::runtime_error( msg.str() );
					}
				}
			}

			--count_down_R;
			if (0 == count_down_R) {
				ptime   time_stop_R = microsec_clock::local_time();
				float   time_last_s = getTimeDuration( time_start_R, time_stop_R );
				float   tps = N_ROUND / time_last_s;
				std::cerr << "  R-GET " << N_ROUND << " records last " << std::setprecision(3) << std::fixed << time_last_s << " seconds, tps=" << tps << std::resetiosflags(std::ios::floatfield) << std::setprecision(6) << std::endl;

				time_start_R = time_stop_R;
				count_down_R = N_ROUND;
			}
		}

		ptime   time_stop = microsec_clock::local_time();
		float   time_last_s = getTimeDuration( time_start, time_stop );
		float   tps = times / time_last_s;
		std::cerr << "GET " << times << " records last " << std::setprecision(3) << std::fixed << time_last_s << " seconds, tps=" << tps << std::resetiosflags(std::ios::floatfield) << std::setprecision(6) << std::endl;
	}



	{
		ptime   time_start = microsec_clock::local_time();
		ptime	time_start_R = time_start;
		u_int32_t	count_down_R = N_ROUND;

		for (int i=0; i<times; i++) {
			char	key[32];
			//char	value[32];

			snprintf( key, sizeof(key) - 1, "%s-KEY-%08d", DBT_PREFIX, i+1 );

            std::string _key( key );
			int rc = clt.erase( _key );
			if (rc < 0) {
				std::ostringstream	msg;
				msg << "fail to ERASE(table:" << table_name << ", key:" << key << ")";
				throw std::runtime_error( msg.str() );
			}

			--count_down_R;
			if (0 == count_down_R) {
				ptime   time_stop_R = microsec_clock::local_time();
				float   time_last_s = getTimeDuration( time_start_R, time_stop_R );
				float   tps = N_ROUND / time_last_s;
				std::cerr << "  R-ERASE " << N_ROUND << " records last " << std::setprecision(3) << std::fixed << time_last_s << " seconds, tps=" << tps << std::resetiosflags(std::ios::floatfield) << std::setprecision(6) << std::endl;

				time_start_R = time_stop_R;
				count_down_R = N_ROUND;
			}
		}

		ptime   time_stop = microsec_clock::local_time();
		float   time_last_s = getTimeDuration( time_start, time_stop );
		float   tps = times / time_last_s;
		std::cerr << "ERASE " << times << " records last " << std::setprecision(3) << std::fixed << time_last_s << " seconds, tps=" << tps << std::resetiosflags(std::ios::floatfield) << std::setprecision(6) << std::endl;
	}

}


int		main(int argc, char* argv[])
{

    std::string appconf;
    //std::string logconf;
    int times = 0;
    int thread_count = 0;
    bool verbose = false;

    try {

        po::options_description desc("Test InsertIgnore/Find/Erase   Allowed options");
        desc.add_options()
            ("help,h", "produce help message")
            ("verbose,v", "output verbose message")
            ("app-conf", po::value<std::string>(&appconf)->default_value("app.conf"), "application config filename")
            //("log-conf", po::value<std::string>(&logconf)->default_value("log.conf"), "log config filename")
            ("times", po::value<int>(&times)->default_value(10), "number of request to send")
            ("threads", po::value<int>(&thread_count)->default_value(1), "number of threads to run")
        ;

        po::variables_map vm;
        po::store(po::parse_command_line(argc, argv, desc), vm);
        po::notify(vm);

        if (vm.count("help")) {
            std::cout << desc << "\n";
            return 1;
        }

        if (vm.count("verbose")) {
            verbose = true;
        }

    }
    catch(std::exception& e) {
        std::cerr << "error: " << e.what() << "\n";
        return 1;
    }
    catch(...) {
        std::cerr << "Exception of unknown type!\n";
        return 2;
    }


    boost::filesystem::path appconf_path( appconf );
    if ( !boost::filesystem::exists( appconf_path ) ) { 
		std::cerr << "config file " << appconf << " does NOT exist.\n";
		return 3;
    }   
    if ( !boost::filesystem::is_regular_file( appconf_path ) ) { 
		std::cerr << "config file " << appconf << " is NOT a regular file.\n";
		return 4;
    }   

	
	/*
    boost::filesystem::path logconf_path( logconf );
    if ( !boost::filesystem::exists( logconf_path ) ) { 
		std::cerr << "config file " << logconf << " does NOT exist.\n";
		return 3;
    }   
    if ( !boost::filesystem::is_regular_file( logconf_path ) ) { 
		std::cerr << "config file " << logconf << " is NOT a regular file.\n";
		return 4;
    }   
	*/
	


    std::vector< boost::thread * >    threads;

    for (int i=0; i<thread_count; i++) {
        boost::thread*   thr = new boost::thread( boost::bind(testfunc, appconf, times, verbose, i) );

        threads.push_back( thr );
    }

    for (size_t i=0; i<threads.size(); i++) {
        boost::thread*  thr = threads[ i ];
        thr->join();
    }

    for (size_t i=0; i<threads.size(); i++) {
        boost::thread*  thr = threads[ i ];
        delete thr;
    }
    threads.clear();


	return	0;
}

